package io.kiki.sba.registry.server.metadata.monitor.impl;

import io.kiki.sba.registry.common.model.metaserver.inter.heartbeat.HeartbeatRequest;
import io.kiki.sba.registry.common.model.metaserver.nodes.DataStoreNode;
import io.kiki.sba.registry.common.model.slot.Slots;
import io.kiki.sba.registry.exception.InitializeException;
import io.kiki.sba.registry.exception.StartException;
import io.kiki.sba.registry.exception.StopException;
import io.kiki.sba.registry.lifecycle.impl.AbstractLifecycle;
import io.kiki.sba.registry.lifecycle.impl.LifecycleHelper;
import io.kiki.sba.registry.observer.Observable;
import io.kiki.sba.registry.observer.UnblockingObserver;
import io.kiki.sba.registry.server.metadata.MetadataLeaderService;
import io.kiki.sba.registry.server.metadata.bootstrap.config.ServerConfig;
import io.kiki.sba.registry.server.metadata.monitor.Metrics;
import io.kiki.sba.registry.server.metadata.monitor.SlotsMonitor;
import io.kiki.sba.registry.server.metadata.monitor.SlotsStats;
import io.kiki.sba.registry.server.metadata.monitor.dataStoreNode.DataSlotMetricsRecorder;
import io.kiki.sba.registry.server.metadata.slot.SlotManager;
import io.kiki.sba.registry.server.shared.resource.SlotGenericResource;
import io.kiki.sba.registry.server.shared.slot.DiskSlotsRecorder;
import io.kiki.sba.registry.server.shared.slot.SlotsRecorder;
import io.kiki.sba.registry.util.ConcurrentUtils;
import io.kiki.sba.registry.util.WakeUpLoopExecuteTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Component
public class SlotsMonitorImpl extends AbstractLifecycle implements SlotsMonitor, UnblockingObserver {
    private static final Logger logger = LoggerFactory.getLogger(SlotsMonitorImpl.class);
    private final Map<String, Integer> dataServerLagCounter = new ConcurrentHashMap<>();
    @Autowired
    private SlotManager slotManager;
    @Autowired
    private SlotGenericResource slotGenericResource;
    @Autowired
    private ServerConfig serverConfig;
    @Autowired
    private MetadataLeaderService metadataLeaderService;
    private SlotsStats SlotsStats;

    private List<SlotsRecorder> recorders;

    private WakeUpLoopExecuteTask scheduledTask;

    @PostConstruct
    public void postConstruct() throws Exception {
        LifecycleHelper.initializeIfPossible(this);
        LifecycleHelper.startIfPossible(this);
    }

    @PreDestroy
    public void preDestroy() throws Exception {
        LifecycleHelper.stopIfPossible(this);
        LifecycleHelper.disposeIfPossible(this);
    }

    @Override
    protected void doInitialize() throws InitializeException {
        super.doInitialize();
        recorders = new ArrayList<>(Arrays.asList(new DiskSlotsRecorder(), slotGenericResource, new DataSlotMetricsRecorder()));
        SlotsStats = new SlotsStatsImpl(slotManager, serverConfig);
        SlotsStats.initialize();
        scheduledTask = new WakeUpLoopExecuteTask() {
            @Override
            public int getWaitingMillis() {
                return 10 * 60 * 1000;
            }

            @Override
            public void _execute_() {
                recordSlots();
            }
        };
    }

    @Override
    protected void doStart() throws StartException {
        super.doStart();
        slotManager.addObserver(this);
        ConcurrentUtils.createDaemonThread(SlotsMonitorImpl.class.getSimpleName(), scheduledTask).start();
    }

    @Override
    protected void doStop() throws StopException {
        slotManager.removeObserver(this);
        if (scheduledTask != null) {
            scheduledTask.close();
        }
        super.doStop();
    }

    @Override
    public void recordSlots() {
        recorders.forEach(recorder -> {
            if (recorder != null) {
                recorder.record(slotManager.getSlots());
            }
        });
    }

    @Override
    public boolean isStableSlots() {
        return SlotsStats.isSlotLeadersStable() && SlotsStats.isSlotFollowersStable();
    }

    @Override
    public void update(Observable source, Object message) {
        if (message instanceof Slots) {
            logger.warn("[update] slot-table changed, current epoch: [{}]", ((Slots) message).getEpoch());
            recordSlots();
            SlotsStats.updateSlots((Slots) message);
        }
    }


    public SlotsMonitorImpl setSlotManager(SlotManager slotManager) {
        this.slotManager = slotManager;
        return this;
    }


    public SlotsMonitorImpl setMetaServerConfig(ServerConfig serverConfig) {
        this.serverConfig = serverConfig;
        return this;
    }

    @Override
    public void onHeartbeat(HeartbeatRequest<DataStoreNode> heartbeatRequest) {
        if (!metadataLeaderService.amIStableAsLeader()) {
            logger.info("[onHeartbeat] I'm not leader or not stable now, do not update slot stats");
            return;
        }
        long slotsEpoch = heartbeatRequest.getSlotsEpoch();
        if (slotsEpoch < slotManager.getSlots().getEpoch()) {
            // after slot-table changed, first time data-server report heartbeat, the epoch is less than
            // meta servers
            // it is ok with that
            // but the second time should be ok, otherwise, data has something un-common
            int times = dataServerLagCounter.getOrDefault(heartbeatRequest.getNode().getIp(), 0) + 1;
            if (times > 1) {
                logger.error("[onHeartbeatLag] data[{}] lag", heartbeatRequest.getNode().getIp());
            }
            Metrics.DataSlot.setDataServerSlotLagTimes(heartbeatRequest.getNode().getIp(), times);
            dataServerLagCounter.put(heartbeatRequest.getNode().getIp(), times);
            return;
        }
        if (heartbeatRequest.getSlotStatuses() == null) {
            logger.warn("[onHeartbeatEmpty] empty heartbeat");
            return;
        }
        dataServerLagCounter.put(heartbeatRequest.getNode().getIp(), 0);
        Metrics.DataSlot.setDataServerSlotLagTimes(heartbeatRequest.getNode().getIp(), 0);
        SlotsStats.checkSlotStatuses(heartbeatRequest.getNode(), heartbeatRequest.getSlotStatuses());
    }


    public SlotsMonitorImpl setMetaLeaderService(MetadataLeaderService metadataLeaderService) {
        this.metadataLeaderService = metadataLeaderService;
        return this;
    }


    public SlotsStats getSlotsStats() {
        return SlotsStats;
    }


    public SlotsMonitorImpl setSlotsStats(SlotsStats SlotsStats) {
        this.SlotsStats = SlotsStats;
        return this;
    }
}
